package com.remoter.transport.socket.client.handler;

import java.net.InetSocketAddress;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import com.google.common.util.concurrent.SettableFuture;
import com.remoter.basic.bean.support.RemoterRequest;
import com.remoter.basic.bean.support.RemoterResponse;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.WriteTimeoutException;

/**
 * @author			koko
 * @date 			2017-08-09 09:36:06
 * @description 	类功能说明
 */
public class ClientMessageHandler extends ChannelInboundHandlerAdapter{
	
	private ConcurrentHashMap<Long,SettableFuture<RemoterResponse>> callbacks = new ConcurrentHashMap<Long,SettableFuture<RemoterResponse>>();
	private volatile Channel channel;
	private volatile InetSocketAddress address;
	
	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception{
		this.channel = ctx.channel();
	}
	
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		this.address = (InetSocketAddress)this.channel.remoteAddress();
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception{
		if(null != msg && msg instanceof RemoterResponse){
			RemoterResponse remoterResponse = (RemoterResponse)msg;
			long token = remoterResponse.getToken();
			if(this.callbacks.containsKey(token)){
				SettableFuture<RemoterResponse> future = this.callbacks.get(token);
				this.callbacks.remove(token);
				future.set(remoterResponse);
			}
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause)throws Exception{
		ctx.close();
	}
	
	public SettableFuture<RemoterResponse> sendRemoterRequest(int timeout,RemoterRequest remoterRequest)throws WriteTimeoutException{
		SettableFuture<RemoterResponse> callback = SettableFuture.create();
		try{
			remoterRequest.setToHost(this.address.getHostString());
			remoterRequest.setToPort(this.address.getPort());
			this.callbacks.put(remoterRequest.getToken(),callback);
			this.channel.writeAndFlush(remoterRequest);
		}catch(Exception e){
			throw WriteTimeoutException.INSTANCE;
		}
		return callback;
	}
	
	public void close()throws Exception{
		for(Entry<Long,SettableFuture<RemoterResponse>> entry : this.callbacks.entrySet()){
			RemoterResponse remoterResponse = new RemoterResponse(entry.getKey());
			remoterResponse.setException("provider closed");
			entry.getValue().set(remoterResponse);
		}
		this.callbacks.clear();
		this.channel.close();
	}
	
	public boolean isAvailable(){
		return this.channel.isOpen();
	}
	
}